#include "local_mqtt.h"
#include "mosquitto.h"
#include "sys_log.h"
#include "thread_msg.h"
#include <string.h>
#include <stdlib.h>

#define MQTT_QOS 1

struct mosquitto* local_mqtt_id;
static int mqtt_connect = 0;
static int mqtt_pub_id = 0, mqtt_pub_calback_id = -1;

void mqtt_log_callback(struct mosquitto *mosq, void *obj, int level, const char *str)
{
	printf("mqtt_log:%s\n", str);
}

static void mqtt_message_callback(struct mosquitto *mosq, void *obj, 
                                     const struct mosquitto_message *msg)
{
    msg_t queue_msg;
    struct mosq_ctrl *ctrl = obj;
    memset(&queue_msg, 0, sizeof(msg_t));
    if (strlen(msg->topic) && strlen(msg->payload)) {
        sys_log_info("mqtt->%s : %s\n", (char*)msg->topic, (char*)msg->payload);
        if (strstr(msg->topic, MQ_RACK_APP) != NULL)
        {
            queue_msg.type = MSG_ID_MQ_APP_RACK;
        } else {
            return;
        }
        queue_msg.topic = (char *)calloc(1, strlen(msg->topic) + 1);
        if (queue_msg.topic == NULL) {
            sys_log_info("calloc failed\n");
            return;
        }
        strcpy(queue_msg.topic, (char *)msg->topic);

        queue_msg.payload = (char *)calloc(1, strlen(msg->payload) + 1);
        if (queue_msg.payload == NULL) {
            sys_log_info("calloc failed\n");
            free(queue_msg.topic);
            return;
        }
        strcpy(queue_msg.payload, msg->payload);
        thread_msg_send(&queue_msg);
    }
}
/**
 * @brief callback function when connection established
 * 
 * @param mosq 
 * @param obj 
 * @param result 
 * @param flags 
 */
static void mqtt_connect_callback(struct mosquitto *mosq, void *obj, int reason_code)
{
    sys_log_info("Mosquitto connect result=%d\n", reason_code);
    if (reason_code == 0) {
        mqtt_connect = 1;
    } else {
        mqtt_connect = 0;
    }
}
/**
 * @brief callback function when connection disconnected
 * 
 * @param mosq 
 * @param obj 
 * @param rc 
 */
static void mqtt_disconnect_callback(struct mosquitto *mosq, void *obj, int rc)
{
    sys_log_info("mqtt disconnected!\n");
    mqtt_connect = 0;
}
/**
 * @brief callback function when publish message
 * 
 * @param mosq 
 * @param obj 
 * @param mid 
 */
static void mqtt_publish_callback(struct mosquitto *mosq, void *obj, int mid)
{
    mqtt_pub_calback_id = mid;
}

static void mqtt_subscribe_callback(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
{
    sys_log_info("success\n");
}

static int local_mqtt_connect(void)
{
    int rc = MOSQ_ERR_NO_CONN;

    if (local_mqtt_id != NULL) {
        mosquitto_loop_stop(local_mqtt_id, true);
        mosquitto_destroy(local_mqtt_id);
        local_mqtt_id = NULL;
    }

    local_mqtt_id = mosquitto_new("sys_manager", true, NULL);
    mosquitto_message_callback_set(local_mqtt_id, &mqtt_message_callback);
    mosquitto_connect_callback_set(local_mqtt_id, &mqtt_connect_callback);
    mosquitto_disconnect_callback_set(local_mqtt_id, &mqtt_disconnect_callback);
    mosquitto_publish_callback_set(local_mqtt_id, &mqtt_publish_callback);
    mosquitto_subscribe_callback_set(local_mqtt_id, &mqtt_subscribe_callback);
    sys_log_info("mosquitto_connect\n");
    rc = mosquitto_connect(local_mqtt_id, "127.0.0.1", 1883, 60);
    sys_log_info("mosquitto_connect\n");
    if (rc != MOSQ_ERR_SUCCESS) {
        sys_log_erro("Mqtt connect to error:%s\n", mosquitto_strerror(rc));
        return -1;
    }
    mosquitto_loop_start(local_mqtt_id);
    while (mqtt_connect == 0) {
        usleep(100000);
        sys_log_erro("Waitting mosquitto connect\n");
    }
    sys_log_info(" mosquitto connected\n");
    return 0;
}

int local_mqtt_pub(const char* topic, const char* data)
{
    int wait = 10;
    int rc = MOSQ_ERR_NO_CONN;

    mqtt_pub_id++;
    sys_log_info("pub topic: %s\n", topic);
    sys_log_info("pub data: %s\n", data);
    rc = mosquitto_publish(local_mqtt_id, &mqtt_pub_id, topic, strlen(data), data, MQTT_QOS, false);

    if (rc != MOSQ_ERR_SUCCESS) {
        sys_log_erro("Publish to %s error:%s\n", topic, mosquitto_strerror(rc));
        return -1;
    }
    while (wait--) {
        if (mqtt_pub_id == mqtt_pub_calback_id) {
            return 0;
        }
        usleep(10);
    }
    return -1;
}

int local_mqtt_sub(const char* topic)
{
    int rc = MOSQ_ERR_NO_CONN;
    if (local_mqtt_id == NULL) {
        sys_log_erro("local_mqtt_id is null\n");
        return -1;
    }
    sys_log_info("sub topic: %s\n", topic);
    rc = mosquitto_subscribe(local_mqtt_id, NULL, topic, MQTT_QOS);
    if (rc != MOSQ_ERR_SUCCESS) {
        sys_log_erro("Publish to %s error:%s\n", topic, mosquitto_strerror(rc));
        return -1;
    }
    return 0;
}

void mosquitto_version(void)
{
	int major, minor, revision;

	mosquitto_lib_version(&major, &minor, &revision);
	sys_log_info("mosquittom running on libmosquitto %d.%d.%d.\n", major, minor, revision);
}

int local_mqtt_init(void)
{
	int ret = 0;
    local_mqtt_id = NULL;

    mosquitto_lib_cleanup();
    mosquitto_lib_init();
    mosquitto_version();

    ret = local_mqtt_connect();
    return ret;
}
